package com.shujia.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.yarn.webapp.view.HtmlPage;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class Demo03HBaseAPI {
    // 定义全局变量，让每一个方法中都能够调用
    Connection conn;
    Admin admin;

    @Before
    public void init() throws IOException {
        System.out.println("执行了init的逻辑");
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");

        conn = ConnectionFactory.createConnection(conf);

        admin = conn.getAdmin();

    }

    @Test
    public void createTable() throws IOException {

        TableName new_t1_table_name = TableName.valueOf("new_t1");

        // 通过tableExists方法判断表是否存在，不存在则创建
        if (admin.tableExists(new_t1_table_name)) {
            System.out.println("new_t1 已经存在！");
        } else {
            HTableDescriptor new_t1 = new HTableDescriptor(TableName.valueOf("new_t1"));
            HColumnDescriptor cf1 = new HColumnDescriptor("cf1");
            cf1.setVersions(3, 10);

            HColumnDescriptor cf2 = new HColumnDescriptor("cf2");
            cf2.setMaxVersions(5);

            new_t1.addFamily(cf1);
            new_t1.addFamily(cf2);

            admin.createTable(new_t1);
        }


    }

    @Test
    public void dropTable() throws IOException {


        TableName new_t1_table_name = TableName.valueOf("new_t1");

        admin.disableTable(new_t1_table_name);
        admin.deleteTable(new_t1_table_name);
    }

    @Test
    public void listTables() throws IOException {
        HTableDescriptor[] hTableDescriptors = admin.listTables();
        for (HTableDescriptor descriptor : hTableDescriptors) {
            System.out.println(descriptor.getNameAsString());
            HColumnDescriptor[] families = descriptor.getColumnFamilies();
            for (HColumnDescriptor cf : families) {
                System.out.println(cf.getNameAsString());
            }
            System.out.println("####################");
        }
    }

    @Test
    public void alterTable() throws IOException {
        TableName new_t1_table_name = TableName.valueOf("new_t1");
        // 要想对表结构进行修改，得先获得原有的结构
        HTableDescriptor tableDescriptor = admin.getTableDescriptor(new_t1_table_name);
        HColumnDescriptor[] columnFamilies = tableDescriptor.getColumnFamilies();
        for (HColumnDescriptor cf : columnFamilies) {
            if ("cf1".equals(cf.getNameAsString())) {
                cf.setVersions(2, 8);
            }
        }

        admin.modifyTable(new_t1_table_name, tableDescriptor);
    }

    @Test
    public void putOneData() throws IOException {
        TableName new_t1_table_name = TableName.valueOf("new_t1");
        Table new_t1 = conn.getTable(new_t1_table_name);
        // 向new_t1表的rowkey为rk1这一行数据的cf1列簇下插入一列数据，列名为id，值为001
        Put put = new Put("rk1".getBytes());
        put.addColumn("cf1".getBytes(), "id".getBytes(), "001".getBytes());
        put.addColumn("cf1".getBytes(), "name".getBytes(), "zhangsan".getBytes());

        new_t1.put(put);


    }

    @Test
    public void getOneData() throws IOException {
        // 获取putOneData方法中写入的数据
        TableName new_t1_table_name = TableName.valueOf("new_t1");
        Table new_t1 = conn.getTable(new_t1_table_name);

        Get get = new Get("rk1".getBytes());
        // 通过rowkey可以获取某一条数据，该条数据中会包含n个cell单元格
        Result res = new_t1.get(get);

        // 获取每一条数据结构都固定的数据
        byte[] id = res.getValue("cf1".getBytes(), "id".getBytes());
        byte[] name = res.getValue("cf1".getBytes(), "name".getBytes());

        // 需要使用HBase提供的Bytes工具 将二进制数组转换成Java中的类型
        System.out.println(Bytes.toString(id));
        System.out.println(Bytes.toString(name));

        // 每一条数据结构不统一时，那如果还是按照上述方法获取数据，会导致有很多空值的出现
        List<Cell> cells = res.listCells();
        for (Cell cell : cells) {
            // 提取当前行数据每一个cell单元格
            String rk = Bytes.toString(CellUtil.cloneRow(cell));
            String cf = Bytes.toString(CellUtil.cloneFamily(cell));
            String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
            String value = Bytes.toString(CellUtil.cloneValue(cell));
            System.out.println(rk + "," + cf + ":" + qualifier + " " + value);

        }


    }

    @Test
    // 读取students.txt数据 写入到HBase的stu表中，以每条数据的id作为rowkey
    // stu表包含info列簇，info列簇分别包含了：id、name、age、gender、clazz五列
    public void putStu() throws IOException {
        TableName stu_table_name = TableName.valueOf("stu");
        // 创建stu表
        if (!admin.tableExists(stu_table_name)) {
            HTableDescriptor hTableDescriptor = new HTableDescriptor(stu_table_name);
            HColumnDescriptor info = new HColumnDescriptor("info");
            hTableDescriptor.addFamily(info);
            admin.createTable(hTableDescriptor);
        }

        Table stu = conn.getTable(stu_table_name);

        // 读取文件，获取每一条数据
        BufferedReader br = new BufferedReader(new FileReader("./data/students.txt"));
        String line;
        while ((line = br.readLine()) != null) {
            String[] splits = line.split(",");
            String id = splits[0];
            String name = splits[1];
            String age = splits[2];
            String gender = splits[3];
            String clazz = splits[4];

            Put put = new Put(id.getBytes());
            put.addColumn("info".getBytes(), "id".getBytes(), id.getBytes());
            put.addColumn("info".getBytes(), "name".getBytes(), name.getBytes());
            put.addColumn("info".getBytes(), "age".getBytes(), age.getBytes());
            put.addColumn("info".getBytes(), "gender".getBytes(), gender.getBytes());
            put.addColumn("info".getBytes(), "clazz".getBytes(), clazz.getBytes());
            stu.put(put);
        }

    }

    @Test
    // 使用批量插入的方式将数据写入HBase
    public void putStuBatch() throws IOException {
        TableName stu_table_name = TableName.valueOf("stu_batch");
        // 创建stu_batch表
        if (!admin.tableExists(stu_table_name)) {
            HTableDescriptor hTableDescriptor = new HTableDescriptor(stu_table_name);
            HColumnDescriptor info = new HColumnDescriptor("info");
            hTableDescriptor.addFamily(info);
            admin.createTable(hTableDescriptor);
        }

        Table stu_batch = conn.getTable(stu_table_name);

        // 创建List接收Put对象
        // 将每一条数据构建成一个Put对象 并添加到puts列表中
        ArrayList<Put> puts = new ArrayList<>();

        // 每100条作为一个批次进行插入
        int batch_size = 88;

        // 读取文件，获取每一条数据
        BufferedReader br = new BufferedReader(new FileReader("./data/students.txt"));
        String line;
        while ((line = br.readLine()) != null) {
            String[] splits = line.split(",");
            String id = splits[0];
            String name = splits[1];
            String age = splits[2];
            String gender = splits[3];
            String clazz = splits[4];

            Put put = new Put(id.getBytes());
            put.addColumn("info".getBytes(), "id".getBytes(), id.getBytes());
            put.addColumn("info".getBytes(), "name".getBytes(), name.getBytes());
            put.addColumn("info".getBytes(), "age".getBytes(), age.getBytes());
            put.addColumn("info".getBytes(), "gender".getBytes(), gender.getBytes());
            put.addColumn("info".getBytes(), "clazz".getBytes(), clazz.getBytes());
            // 将put对象添加到List中
            puts.add(put);

            if (puts.size() == batch_size) {
                // 进行批量的写入
                stu_batch.put(puts);
                // 对List进行清空
                puts.clear();
            }
        }

        // 判断List中是否还有数据未写入
        if (puts.size() > 0) {
            stu_batch.put(puts);
        }


    }

    @Test
    public void delete() throws IOException {
        TableName t1_table_name = TableName.valueOf("t1");
        Table t1 = conn.getTable(t1_table_name);

        Delete delete = new Delete("rk1".getBytes());

        t1.delete(delete);

    }

    @Test
    public void scan() throws IOException {
        TableName stu_table_name = TableName.valueOf("stu");
        Table stu = conn.getTable(stu_table_name);
        // 返回rowkey范围在1500100888~1500100900的前10条数据
        Scan scan = new Scan();
        scan.setLimit(10);
        scan.withStartRow("1500100888".getBytes());
        scan.withStopRow("1500100900".getBytes());


        ResultScanner resultScanner = stu.getScanner(scan);
        for (Result rs : resultScanner) {
            String id = Bytes.toString(rs.getValue("info".getBytes(), "id".getBytes()));
            String name = Bytes.toString(rs.getValue("info".getBytes(), "name".getBytes()));
            String age = Bytes.toString(rs.getValue("info".getBytes(), "age".getBytes()));
            String gender = Bytes.toString(rs.getValue("info".getBytes(), "gender".getBytes()));
            String clazz = Bytes.toString(rs.getValue("info".getBytes(), "clazz".getBytes()));

            System.out.println(id + "," + name + "," + age + "," + gender + "," + clazz);
        }

    }

    @Test
    // 读取DIANXIN.csv数据 写入到HBase的dianxin表中，以每条数据的mdn作为rowkey
    // dianxin表包含一个列簇info，info包含两列数据：lg经度、lat维度
    // 以起始时间(第三列)作为时间戳
    // 能够基于给定的mdn获取用户最新的3条位置记录
    public void putDianXin() throws IOException {
        // 创建dianxin表
        TableName dianxin_t_name = TableName.valueOf("dianxin");
//        if (!admin.tableExists(dianxin_t_name)) {
//            HTableDescriptor hTableDescriptor = new HTableDescriptor(dianxin_t_name);
//            HColumnDescriptor info = new HColumnDescriptor("info");
//            info.setMaxVersions(5);
//            hTableDescriptor.addFamily(info);
//            admin.createTable(hTableDescriptor);
//        }

        Table dianxin = conn.getTable(dianxin_t_name);

        // 读取DIANXIN.csv数据文件
        BufferedReader br = new BufferedReader(new FileReader("./data/DIANXIN.csv"));
        String line;

        ArrayList<Put> puts = new ArrayList<>();
        int batchSize = 1000;

        while ((line = br.readLine()) != null) {
            String[] splits = line.split(",");
            String mdn = splits[0];
            String start_time = splits[2];
            String lg = splits[4];
            String lat = splits[5];
            Put put = new Put(mdn.getBytes());

            put.addColumn("info".getBytes(), "lg".getBytes(), Long.parseLong(start_time), lg.getBytes());
            put.addColumn("info".getBytes(), "lat".getBytes(), Long.parseLong(start_time), lat.getBytes());

            puts.add(put);
            if (puts.size() == batchSize) {
                dianxin.put(puts);
                puts.clear();
            }
        }

        if (puts.size() > 0) {
            dianxin.put(puts);
        }

    }

    @Test
    // 基于给定的mdn作为rowkey，通过get方式，获取用户最新的3条位置记录
    public void getMdnPosition() throws IOException {
        String mdn = "47BE1E866CFC071DB19D5E1C056BE28AE24C16E7";
        TableName dianxin_t_name = TableName.valueOf("dianxin");
        Table dianxin = conn.getTable(dianxin_t_name);

        Get get = new Get(mdn.getBytes());
        get.setMaxVersions(3);

        Result rs = dianxin.get(get);

        // 如果直接通过列簇+列名直接取数据 最终还是会返回一条最新的记录
//        String lg = Bytes.toString(rs.getValue("info".getBytes(), "lg".getBytes()));
//        String lat = Bytes.toString(rs.getValue("info".getBytes(), "lat".getBytes()));
//
//        System.out.println(lg + "," + lat);
        ArrayList<String> lats = new ArrayList<>();
        ArrayList<String> lgs = new ArrayList<>();
        for (Cell cell : rs.listCells()) {
            String rowkey = Bytes.toString(CellUtil.cloneRow(cell));
            String info = Bytes.toString(CellUtil.cloneFamily(cell));
            String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
            String value = Bytes.toString(CellUtil.cloneValue(cell));
            if ("lat".equals(qualifier)) {
                lats.add(value);
            } else if ("lg".equals(qualifier)) {
                lgs.add(value);
            }
        }

        // 格式化输出
        for (int i = 0; i < 3; i++) {
            String lg = lgs.get(i);
            String lat = lats.get(i);
            System.out.println(lg + "," + lat);
        }


    }


    @After
    public void close() throws IOException {
        admin.close();
        conn.close();
    }

}
